package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.app.func.DimAsyncFunction;
import com.atguigu.edu.bean.VideoLookProcessBean;
import com.atguigu.edu.util.ClickHouseUtil;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.time.ZoneId;
import java.util.concurrent.TimeUnit;

/**
 * @author zzw
 * @create 2022/9/4 10:35
 */
public class DwsVideoLookProcessWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        env.setStateBackend(new HashMapStateBackend());
        */

        // 设置时区的方法
        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));

        // TODO 3 读取DWD层 dwd_video_look_process 主题数据
        String topicName = "dwd_traffic_app_Video";
        String groupId = "dws_video_look_process_window";
        DataStreamSource<String> lookProcessStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));

        // TODO 4 过滤出播放视频数据(play_sec != null)   +  转换数据类型
        SingleOutputStreamOperator<JSONObject> jsonObjStream = lookProcessStream.map(JSON::parseObject);

        // TODO 5 根据uid分组

        KeyedStream<JSONObject, String> keyedStream = jsonObjStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getJSONObject("common").getString("uid");
            }
        });

        SingleOutputStreamOperator<VideoLookProcessBean> resultStream = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, VideoLookProcessBean>() {
            ValueState<String> lastLookDtState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                lastLookDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>("last_look_dt", String.class));
            }

            @Override
            public void flatMap(JSONObject value, Collector<VideoLookProcessBean> out) throws Exception {
                String lastLookDt = lastLookDtState.value();
                JSONObject common = value.getJSONObject("common");
                String uid = common.getString("uid");
                Long playSec = value.getLong("play_sec");
                Long ts = value.getLong("ts");
                String videoId = value.getString("video_id");
                String lookDt = DateFormatUtil.toDate(ts);
                if (lastLookDt == null) {
                    //第一条数据
                    out.collect(VideoLookProcessBean.builder()
                            .videoLookCount(0L)
                            .totalTimeCount(playSec)
                            .user_id(uid)
                            .video_id(videoId)
                            .ts(ts)
                            .build());
                    lastLookDtState.update(lookDt);
                } else {
                    if (playSec != 30) {
                        //一次播放
                        out.collect(VideoLookProcessBean.builder()
                                .videoLookCount(1L)
                                .totalTimeCount(playSec)
                                .user_id(uid)
                                .video_id(videoId)
                                .ts(ts)
                                .build());
                    } else {
                        out.collect(VideoLookProcessBean.builder()
                                .videoLookCount(0L)
                                .totalTimeCount(playSec)
                                .user_id(uid)
                                .video_id(videoId)
                                .ts(ts)
                                .build());
                    }
                }
            }
        });
        //resultStream.print("聚合前>>>>>>");
        // TODO 6 开窗聚合
        SingleOutputStreamOperator<VideoLookProcessBean> beanwithWaterMark = resultStream.assignTimestampsAndWatermarks(WatermarkStrategy.<VideoLookProcessBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<VideoLookProcessBean>() {
            @Override
            public long extractTimestamp(VideoLookProcessBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        SingleOutputStreamOperator<VideoLookProcessBean> reduceStream = beanwithWaterMark.keyBy(VideoLookProcessBean::getVideo_id).window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<VideoLookProcessBean>() {
                    @Override
                    public VideoLookProcessBean reduce(VideoLookProcessBean value1, VideoLookProcessBean value2) throws Exception {
                        value1.setVideoLookCount(value1.getVideoLookCount() + value2.getVideoLookCount());
                        value1.setTotalTimeCount(value1.getTotalTimeCount() + value2.getTotalTimeCount());
                        return value1;
                    }
                }, new WindowFunction<VideoLookProcessBean, VideoLookProcessBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<VideoLookProcessBean> input, Collector<VideoLookProcessBean> out) throws Exception {
                        VideoLookProcessBean videoLookProcessBean = input.iterator().next();
                        videoLookProcessBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        videoLookProcessBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        videoLookProcessBean.setTs(System.currentTimeMillis()*1000L);
                        out.collect(videoLookProcessBean);
                    }
                });

        //reduceStream.print("reduce>>>>>>");
        // TODO 7 关联维度表数据
        //关联video_info
        SingleOutputStreamOperator<VideoLookProcessBean> videoInfoStream= AsyncDataStream.unorderedWait(reduceStream,new DimAsyncFunction<VideoLookProcessBean>("DIM_VIDEO_INFO"){
@Override
public String getKey(VideoLookProcessBean input){
        return input.getVideo_id();
        }

@Override
public void join(VideoLookProcessBean input,JSONObject obj){
        input.setVideo_name(obj.getString("videoName"));
        input.setChapter_id(obj.getString("chapterId"));
        input.setCourse_id(obj.getString("courseId"));
        }
        },100, TimeUnit.SECONDS);

        //关联course_info表
        SingleOutputStreamOperator<VideoLookProcessBean> courseNameStream=AsyncDataStream.unorderedWait(videoInfoStream,new DimAsyncFunction<VideoLookProcessBean>("DIM_COURSE_INFO"){
@Override
public String getKey(VideoLookProcessBean input){
        return input.getChapter_id();
        }

@Override
public void join(VideoLookProcessBean input,JSONObject obj){
        input.setCourse_name(obj.getString("courseName"));

        }
        },100,TimeUnit.SECONDS);

        courseNameStream.print("写入ClickHouse>>>>>>");

        courseNameStream.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into dws_video_look_process_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();

    }
}
